package top.reminis.conf;

import lombok.extern.slf4j.Slf4j;
import org.yaml.snakeyaml.Yaml;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 作业配置加载器
 * 支持从YAML文件加载作业配置
 * 
 * @author zwf
 * @create 2025-09-24
 */
@Slf4j
public class JobConfigLoader {

    /**
     * 加载作业配置
     */
    public static JobConfig load(String configFile) {
        try {
            Yaml yaml = new Yaml();
            InputStream inputStream = JobConfigLoader.class
                    .getClassLoader()
                    .getResourceAsStream(configFile);
            
            if (inputStream == null) {
                throw new RuntimeException("配置文件不存在: " + configFile);
            }
            
            Map<String, Object> configMap = yaml.load(inputStream);
            return parseConfig(configMap);
            
        } catch (Exception e) {
            log.error("加载配置文件失败: {}", configFile, e);
            throw new RuntimeException("加载配置文件失败: " + configFile, e);
        }
    }

    /**
     * 解析配置Map为JobConfig对象
     */
    @SuppressWarnings("unchecked")
    private static JobConfig parseConfig(Map<String, Object> configMap) {
        JobConfig config = new JobConfig();
        
        // 解析作业基本信息
        Map<String, Object> jobInfo = (Map<String, Object>) configMap.get("job");
        if (jobInfo != null) {
            config.setJobName((String) jobInfo.get("name"));
            config.setDescription((String) jobInfo.get("description"));
            config.setVersion((String) jobInfo.get("version"));
        }
        
        // 解析Flink执行配置
        Map<String, Object> flinkConfig = (Map<String, Object>) configMap.get("flink");
        if (flinkConfig == null) {
            throw new IllegalStateException("缺少 Flink 配置，请在配置文件中添加 flink 配置段");
        }
        
        config.setParallelism((Integer) flinkConfig.get("parallelism"));
        config.setCheckpointInterval(((Number) flinkConfig.get("checkpoint_interval")).longValue());
        config.setCheckpointTimeout(((Number) flinkConfig.get("checkpoint_timeout")).longValue());
        config.setMinPauseBetweenCheckpoints(((Number) flinkConfig.get("min_pause_between_checkpoints")).longValue());
        config.setMaxConcurrentCheckpoints((Integer) flinkConfig.get("max_concurrent_checkpoints"));
        config.setTolerableCheckpointFailureNumber((Integer) flinkConfig.get("tolerable_checkpoint_failure_number"));
        config.setMaxRestartAttempts((Integer) flinkConfig.get("max_restart_attempts"));
        config.setRestartDelaySeconds((Integer) flinkConfig.get("restart_delay_seconds"));
        
        // 解析监控配置
        Map<String, Object> monitoringConfig = (Map<String, Object>) configMap.get("monitoring");
        if (monitoringConfig == null) {
            throw new IllegalStateException("缺少监控配置，请在配置文件中添加 monitoring 配置段");
        }
        
        config.setMonitoringInterval((Integer) monitoringConfig.get("interval"));
        config.setLogLevel((String) monitoringConfig.get("log_level"));
        config.setEnableDebugSql((Boolean) monitoringConfig.get("enable_debug_sql"));
        
        // 解析数据源配置
        List<Map<String, Object>> dataSourcesList = (List<Map<String, Object>>) configMap.get("data_sources");
        if (dataSourcesList != null) {
            List<JobConfig.DataSourceConfig> dataSources = new ArrayList<>();
            for (Map<String, Object> sourceMap : dataSourcesList) {
                dataSources.add(parseDataSourceConfig(sourceMap));
            }
            config.setDataSources(dataSources);
        }
        
        // 解析数据目标配置
        List<Map<String, Object>> dataSinksList = (List<Map<String, Object>>) configMap.get("data_sinks");
        if (dataSinksList != null) {
            List<JobConfig.DataSinkConfig> dataSinks = new ArrayList<>();
            for (Map<String, Object> sinkMap : dataSinksList) {
                dataSinks.add(parseDataSinkConfig(sinkMap));
            }
            config.setDataSinks(dataSinks);
        }
        
        // 解析数据流配置
        List<Map<String, Object>> dataFlowsList = (List<Map<String, Object>>) configMap.get("data_flows");
        if (dataFlowsList != null) {
            List<JobConfig.DataFlowConfig> dataFlows = new ArrayList<>();
            for (Map<String, Object> flowMap : dataFlowsList) {
                dataFlows.add(parseDataFlowConfig(flowMap));
            }
            config.setDataFlows(dataFlows);
        }
        
        // 解析Catalog配置
        List<Map<String, Object>> catalogsList = (List<Map<String, Object>>) configMap.get("catalogs");
        if (catalogsList != null) {
            List<JobConfig.CatalogConfig> catalogs = new ArrayList<>();
            for (Map<String, Object> catalogMap : catalogsList) {
                catalogs.add(parseCatalogConfig(catalogMap));
            }
            config.setCatalogs(catalogs);
        }
        
        // 解析存储配置
        Map<String, Object> storageMap = (Map<String, Object>) configMap.get("storage");
        if (storageMap != null) {
            config.setStorage(parseStorageConfig(storageMap));
        }
        
        // 解析表属性配置
        Map<String, Object> tableProps = (Map<String, Object>) configMap.get("table_properties");
        if (tableProps != null) {
            Map<String, String> tableProperties = new HashMap<>();
            tableProps.forEach((key, value) -> tableProperties.put(key, String.valueOf(value)));
            config.setTableProperties(tableProperties);
        }
        
        return config;
    }

    @SuppressWarnings("unchecked")
    private static JobConfig.DataSourceConfig parseDataSourceConfig(Map<String, Object> sourceMap) {
        JobConfig.DataSourceConfig source = new JobConfig.DataSourceConfig();
        source.setName((String) sourceMap.get("name"));
        source.setType((String) sourceMap.get("type"));
        source.setConnectionString((String) sourceMap.get("connection_string"));
        source.setTopic((String) sourceMap.get("topic"));
        source.setBootstrapServers((String) sourceMap.get("bootstrap_servers"));
        source.setGroupId((String) sourceMap.get("group_id"));
        source.setUsername((String) sourceMap.get("username"));
        source.setPassword((String) sourceMap.get("password"));
        source.setDatabase((String) sourceMap.get("database"));
        source.setTable((String) sourceMap.get("table"));
        
        Map<String, Object> props = (Map<String, Object>) sourceMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            source.setProperties(properties);
        }
        
        // 解析字段定义
        List<Map<String, Object>> fieldsList = (List<Map<String, Object>>) sourceMap.get("fields");
        if (fieldsList != null) {
            List<JobConfig.FieldDefinition> fields = new ArrayList<>();
            for (Map<String, Object> fieldMap : fieldsList) {
                fields.add(parseFieldDefinition(fieldMap));
            }
            source.setFields(fields);
        }
        
        return source;
    }

    @SuppressWarnings("unchecked")
    private static JobConfig.DataSinkConfig parseDataSinkConfig(Map<String, Object> sinkMap) {
        JobConfig.DataSinkConfig sink = new JobConfig.DataSinkConfig();
        sink.setName((String) sinkMap.get("name"));
        sink.setType((String) sinkMap.get("type"));
        sink.setConnectionString((String) sinkMap.get("connection_string"));
        sink.setTableName((String) sinkMap.get("table_name"));
        sink.setTopic((String) sinkMap.get("topic"));
        sink.setBootstrapServers((String) sinkMap.get("bootstrap_servers"));
        sink.setUsername((String) sinkMap.get("username"));
        sink.setPassword((String) sinkMap.get("password"));
        sink.setDatabase((String) sinkMap.get("database"));
        
        Map<String, Object> props = (Map<String, Object>) sinkMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            sink.setProperties(properties);
        }
        
        // 解析字段定义
        List<Map<String, Object>> fieldsList = (List<Map<String, Object>>) sinkMap.get("fields");
        if (fieldsList != null) {
            List<JobConfig.FieldDefinition> fields = new ArrayList<>();
            for (Map<String, Object> fieldMap : fieldsList) {
                fields.add(parseFieldDefinition(fieldMap));
            }
            sink.setFields(fields);
        }
        
        return sink;
    }

    @SuppressWarnings("unchecked")
    private static JobConfig.DataFlowConfig parseDataFlowConfig(Map<String, Object> flowMap) {
        JobConfig.DataFlowConfig flow = new JobConfig.DataFlowConfig();
        flow.setName((String) flowMap.get("name"));
        flow.setSourceTable((String) flowMap.get("source_table"));
        flow.setTargetTable((String) flowMap.get("target_table"));
        flow.setTransformationType((String) flowMap.get("transformation_type"));
        flow.setFilterCondition((String) flowMap.get("filter_condition"));
        
        Map<String, Object> rules = (Map<String, Object>) flowMap.get("transformation_rules");
        if (rules != null) {
            Map<String, String> transformationRules = new HashMap<>();
            rules.forEach((key, value) -> transformationRules.put(key, String.valueOf(value)));
            flow.setTransformationRules(transformationRules);
        }
        
        Map<String, Object> props = (Map<String, Object>) flowMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            flow.setProperties(properties);
        }
        
        // 解析字段映射
        List<Map<String, Object>> fieldMappingsList = (List<Map<String, Object>>) flowMap.get("field_mappings");
        if (fieldMappingsList != null) {
            List<JobConfig.FieldMapping> fieldMappings = new ArrayList<>();
            for (Map<String, Object> mappingMap : fieldMappingsList) {
                fieldMappings.add(parseFieldMapping(mappingMap));
            }
            flow.setFieldMappings(fieldMappings);
        }
        
        return flow;
    }

    @SuppressWarnings("unchecked")
    private static JobConfig.CatalogConfig parseCatalogConfig(Map<String, Object> catalogMap) {
        JobConfig.CatalogConfig catalog = new JobConfig.CatalogConfig();
        catalog.setName((String) catalogMap.get("name"));
        catalog.setType((String) catalogMap.get("type"));
        catalog.setConnectionString((String) catalogMap.get("connection_string"));
        catalog.setUri((String) catalogMap.get("uri"));
        catalog.setUsername((String) catalogMap.get("username"));
        catalog.setPassword((String) catalogMap.get("password"));
        catalog.setWarehouse((String) catalogMap.get("warehouse"));
        catalog.setDefault((Boolean) catalogMap.getOrDefault("is_default", false));
        
        Map<String, Object> props = (Map<String, Object>) catalogMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            catalog.setProperties(properties);
        }
        
        return catalog;
    }

    @SuppressWarnings("unchecked")
    private static JobConfig.StorageConfig parseStorageConfig(Map<String, Object> storageMap) {
        JobConfig.StorageConfig storage = new JobConfig.StorageConfig();
        storage.setType((String) storageMap.get("type"));
        storage.setEndpoint((String) storageMap.get("endpoint"));
        storage.setAccessKey((String) storageMap.get("access_key"));
        storage.setSecretKey((String) storageMap.get("secret_key"));
        storage.setBucket((String) storageMap.get("bucket"));
        storage.setRegion((String) storageMap.get("region"));
        storage.setPathStyleAccess((String) storageMap.getOrDefault("path_style_access", "true"));
        storage.setSslEnabled((String) storageMap.getOrDefault("ssl_enabled", "false"));
        
        Map<String, Object> props = (Map<String, Object>) storageMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            storage.setProperties(properties);
        }
        
        return storage;
    }
    
    @SuppressWarnings("unchecked")
    private static JobConfig.FieldDefinition parseFieldDefinition(Map<String, Object> fieldMap) {
        JobConfig.FieldDefinition field = new JobConfig.FieldDefinition();
        field.setName((String) fieldMap.get("name"));
        field.setType((String) fieldMap.get("type"));
        field.setComment((String) fieldMap.get("comment"));
        field.setNotNull((Boolean) fieldMap.getOrDefault("not_null", false));
        field.setPrimaryKey((Boolean) fieldMap.getOrDefault("primary_key", false));
        field.setDefaultValue((String) fieldMap.get("default_value"));
        
        Map<String, Object> props = (Map<String, Object>) fieldMap.get("properties");
        if (props != null) {
            Map<String, String> properties = new HashMap<>();
            props.forEach((key, value) -> properties.put(key, String.valueOf(value)));
            field.setProperties(properties);
        }
        
        return field;
    }
    
    @SuppressWarnings("unchecked")
    private static JobConfig.FieldMapping parseFieldMapping(Map<String, Object> mappingMap) {
        JobConfig.FieldMapping mapping = new JobConfig.FieldMapping();
        mapping.setSource((String) mappingMap.get("source"));
        mapping.setTarget((String) mappingMap.get("target"));
        mapping.setTransform((String) mappingMap.get("transform"));
        mapping.setComment((String) mappingMap.get("comment"));
        return mapping;
    }
}
